#ifndef DIO_IOSTATS_H_
#define DIO_IOSTATS_H_

#ifndef DIO_STATS_TIMING
 #define DIO_STATS_TIMING 0
 #define DO_NOT_COUNT_WAIT_TIME 1
#endif

#include <fermat/common/singleton.h>
#include <fermat/common/timestamp.h>
#include <fermat/common/timespan.h>
#include <fermat/common/mutex.h>

#include <ostream>
#include <string>

namespace fermat {
 
class Stats : public Singleton<Stats> {
    friend class Singleton<Stats>;

    //! number of operations
    size_t _read_ops;
    size_t _write_ops;
    //! number of bytes read/written
    int64_t _read_volume;
	int64_t _write_volume;
    //! number of cached operations
    size_t _cached_read_ops; 
	size_t _cached_write_ops;
    //! number of bytes read/written from/to cache
    int64_t _cached_read_volume;
	int64_t _cached_write_volume;
    //! seconds spent in operations
    Timespan _read_time;
	Timespan _write_time;
    //! seconds spent in parallel operations
    Timespan _parallel_read_time;
	Timespan _parallel_write_time;
    //! seconds spent in all parallel I/O operations (read and write)
    Timespan _parallel_io_time;
    //! seconds spent waiting for completion of I/O operations
    Timespan _io_wait_time;
    Timespan _read_wait_time;
	Timespan _write_wait_time;
#if DIO_STATS_TIMING
    //! start time of parallel operation
    Timestamp _parallel_read_begin;
	Timestamp _parallel_write_begin;
    Timestamp _parallel_io_begin;
    Timestamp _parallel_wait_begin;
    Timestamp _parallel_wait_read_begin;
    Timestamp _parallel_wait_write_begin;
    Timestamp _parallel_wait_time;
    Timestamp _parallel_wait_read_time;
    Timestamp _parallel_wait_write_time;
    //! number of requests, participating in parallel operation
    int _acc_reads;
    int _acc_writes;
    int _acc_ios;
    int _acc_waits;
    int _acc_wait_read;
	int _acc_wait_write;
#endif
    Timestamp _last_reset_time;

    Mutex _read_mutex;
	Mutex _write_mutex;
	Mutex _io_mutex;
	Mutex _wait_mutex;

    Stats();

public:
    enum  WaitOp {
        WOP_ANY, WOP_READ, WOP_WRITE
    };

    class ScopedReadWriteTimer
    {
        typedef size_t  size_type;

        bool _is_write;
        bool _running;

    public:
        explicit ScopedReadWriteTimer(size_type size, bool is_write = false)
            : _is_write(is_write),_running(false){
            Start(size);
        }

        ~ScopedReadWriteTimer() {
            Stop();
        }

        void Start(size_type size) {
            if (!_running) {
                _running = true;
                if (_is_write)
                    Stats::getInstance()->write_started(size);
                else
                    Stats::getInstance()->read_started(size);
            }
        }

        void Stop() {
            if (_running) {
                if (_is_write)
                    Stats::getInstance()->write_finished();
                else
                    Stats::getInstance()->read_finished();
                _running = false;
            }
        }
    };

    class ScopedWriteTimer
    {
        typedef size_t  size_type;
        bool _running;

    public:
        explicit ScopedWriteTimer(size_type size):_running(false) {
            Start(size);
        }

        ~ScopedWriteTimer() {
            Stop();
        }

        void Start(size_type size) {
            if (!_running) {
                _running = true;
                Stats::getInstance()->write_started(size);
            }
        }

        void Stop() {
            if (_running) {
                Stats::getInstance()->write_finished();
                _running = false;
            }
        }
    };

    class ScopedReadTimer
    {
        typedef size_t  size_type;

        bool _running;

    public:
        explicit ScopedReadTimer(size_type size):_running(false) {
            Start(size);
        }

        ~ScopedReadTimer() {
            Stop();
        }

        void Start(size_type size) {
            if (!_running) {
                _running = true;
                Stats::getInstance()->read_started(size);
            }
        }

        void Stop() {
            if (_running) {
                Stats::getInstance()->read_finished();
                _running = false;
            }
        }
    };

    class ScopedWaitTimer
    {
#if !DO_NOT_COUNT_WAIT_TIME
        bool _running = false;
        WaitOp _wait_op;
#endif

    public:
        explicit ScopedWaitTimer(WaitOp wait_op, bool measure_time = true)
#if !DO_NOT_COUNT_WAIT_TIME
            : _wait_op(wait_op)
#endif
        {
            if (measure_time)
                Start();
#if DO_NOT_COUNT_WAIT_TIME
           (void)wait_op;
#endif
        }

        ~ScopedWaitTimer() {
            Stop();
        }

        void Start() {
#if !DO_NOT_COUNT_WAIT_TIME
            if (!_running) {
                _running = true;
                Stats::getInstance()->wait_started(_wait_op);
            }
#endif
        }

        void Stop() {
#if !DO_NOT_COUNT_WAIT_TIME
            if (_running) {
                Stats::getInstance()->wait_finished(_wait_op);
                _running = false;
            }
#endif
        }
    };

public:
    //! Returns total number of read_ops.
    size_t read_ops() const { return _read_ops; }

    //! Returns total number of write_ops.
    size_t write_ops() const { return _write_ops; }

    //! Returns number of bytes read from disks.
    int64_t read_volume() const { return _read_volume; }

    //! Returns number of bytes written to the disks.
    int64_t write_volume() const { return _write_volume; }

    //! Returns total number of reads served from cache.
    size_t cached_read_ops() const { return _cached_read_ops; }

    //! Returns total number of cached write_ops.
    size_t cached_write_ops() const { return _cached_read_ops; }

    //! Returns number of bytes read from cache.
    int64_t cached_read_volume() const { return _cached_read_volume; }

    //! Returns number of bytes written to the cache.
    int64_t cached_write_volume() const { return _cached_write_volume; }

    //! Time that would be spent in read syscalls if all parallel read_ops were
    //! serialized.
    Timespan read_time() const { return _read_time; }

    //! Time that would be spent in write syscalls if all parallel write_ops were
    //! serialized.
    Timespan write_time() const { return _write_time; }

    //! Period of time when at least one I/O thread was executing a read.
    Timespan parallel_read_time() const { return _parallel_read_time; }

    //! Period of time when at least one I/O thread was executing a write.
    Timespan parallel_write_time() const { return _parallel_write_time; }

    //! Period of time when at least one I/O thread was executing a read or a
    //! write.
    Timespan parallel_io_time() const { return _parallel_io_time; }

    //! I/O wait time counter.
    Timespan io_wait_time() const { return _io_wait_time; }

    Timespan read_wait_time() const { return _read_wait_time; }

    Timespan write_wait_time() const { return _write_wait_time; }

    //! Return time of the last reset.
    Timestamp last_reset_time() const { return _last_reset_time; }

    // for library use
    void write_started(size_t size, Timestamp now = Timestamp());
    void write_canceled(size_t size);
    void write_finished();
    void write_cached(size_t size);
    void read_started(size_t size, Timestamp now = Timestamp());
    void read_canceled(size_t size);
    void read_finished();
    void read_cached(size_t size);
    void wait_started(WaitOp wait_op);
    void wait_finished(WaitOp wait_op);
};

#ifdef DO_NOT_COUNT_WAIT_TIME
inline void Stats::wait_started(WaitOp) { }
inline void Stats::wait_finished(WaitOp) { }
#endif

class StatsData
{
    //! number of operations
    size_t _read_ops;
	size_t _write_ops;
    //! number of bytes read/written
    int64_t _read_volume;
	int64_t _write_volume;
    //! number of cached operations
    size_t _cached_read_ops;
	size_t _cached_write_ops;
    //! number of bytes read/written from/to cache
    int64_t _cached_read_volume;
	int64_t _cached_write_volume;
    //! seconds spent in operations
    Timespan  _read_time;
	Timespan   _write_time;
    //! seconds spent in parallel operations
    Timespan  _parallel_read_time;
	Timespan  _parallel_write_time;
    //! seconds spent in all parallel I/O operations (read and write)
    Timespan  _parallel_io_time;
    //! seconds spent waiting for completion of I/O operations
    Timespan  _io_wait_time;
    Timespan  _read_wait_time;
	Timespan  _write_wait_time;
    Timespan  _elapsed_time;

public:
    StatsData()
		: _read_ops(0),
          _write_ops(0),
          _read_volume(0),
          _write_volume(0),
          _cached_read_ops(0),
          _cached_write_ops(0),
          _cached_read_volume(0),
          _cached_write_volume(0)
    {

	}

    explicit StatsData(const Stats& s)
        : _read_ops(s.read_ops()),
          _write_ops(s.write_ops()),
          _read_volume(s.read_volume()),
          _write_volume(s.write_volume()),
          _cached_read_ops(s.cached_read_ops()),
          _cached_write_ops(s.cached_write_ops()),
          _cached_read_volume(s.cached_read_volume()),
          _cached_write_volume(s.cached_write_volume()),
          _read_time(s.read_time()),
          _write_time(s.write_time()),
          _parallel_read_time(s.parallel_read_time()),
          _parallel_write_time(s.parallel_write_time()),
          _parallel_io_time(s.parallel_io_time()),
          _io_wait_time(s.io_wait_time()),
          _read_wait_time(s.read_wait_time()),
          _write_wait_time(s.write_wait_time()),
          _elapsed_time(Timestamp() - s.last_reset_time())
    { }

    StatsData operator + (const StatsData& a) const {
        StatsData s;
        s._read_ops = _read_ops + a._read_ops;
        s._write_ops = _write_ops + a._write_ops;
        s._read_volume = _read_volume + a._read_volume;
        s._write_volume = _write_volume + a._write_volume;
        s._cached_read_ops = _cached_read_ops + a._cached_read_ops;
        s._cached_write_ops = _cached_write_ops + a._cached_write_ops;
        s._cached_read_volume = _cached_read_volume + a._cached_read_volume;
        s._cached_write_volume = _cached_write_volume + a._cached_write_volume;
        s._read_time = _read_time + a._read_time;
        s._write_time = _write_time + a._write_time;
        s._parallel_read_time = _parallel_read_time + a._parallel_read_time;
        s._parallel_write_time = _parallel_write_time + a._parallel_write_time;
        s._parallel_io_time = _parallel_io_time + a._parallel_io_time;
        s._io_wait_time = _io_wait_time + a._io_wait_time;
        s._read_wait_time = _read_wait_time + a._read_wait_time;
        s._write_wait_time = _write_wait_time + a._write_wait_time;
        s._elapsed_time = _elapsed_time + a._elapsed_time;
        return s;
    }

    StatsData operator - (const StatsData& a) const {
        StatsData s;
        s._read_ops = _read_ops - a._read_ops;
        s._write_ops = _write_ops - a._write_ops;
        s._read_volume = _read_volume - a._read_volume;
        s._write_volume = _write_volume - a._write_volume;
        s._cached_read_ops = _cached_read_ops - a._cached_read_ops;
        s._cached_write_ops = _cached_write_ops - a._cached_write_ops;
        s._cached_read_volume = _cached_read_volume - a._cached_read_volume;
        s._cached_write_volume = _cached_write_volume - a._cached_write_volume;
        s._read_time = _read_time - a._read_time;
        s._write_time = _write_time - a._write_time;
        s._parallel_read_time = _parallel_read_time - a._parallel_read_time;
        s._parallel_write_time = _parallel_write_time - a._parallel_write_time;
        s._parallel_io_time = _parallel_io_time - a._parallel_io_time;
        s._io_wait_time = _io_wait_time - a._io_wait_time;
        s._read_wait_time = _read_wait_time - a._read_wait_time;
        s._write_wait_time = _write_wait_time - a._write_wait_time;
        s._elapsed_time = _elapsed_time - a._elapsed_time;
        return s;
    }

    size_t read_ops() const {
        return _read_ops;
    }

    size_t write_ops() const {
        return _write_ops;
    }

    int64_t read_volume() const {
        return _read_volume;
    }

    int64_t write_volume() const {
        return _write_volume;
    }

    size_t _cachedread_ops() const {
        return _cached_read_ops;
    }

    size_t _cachedwrite_ops() const {
        return _cached_write_ops;
    }

    int64_t _cachedread_volume() const {
        return _cached_read_volume;
    }

    int64_t _cachedwrite_volume() const {
        return _cached_write_volume;
    }

    Timespan read_time() const {
        return _read_time;
    }

    Timespan write_time() const {
        return _write_time;
    }

    Timespan parallel_read_time() const {
        return _parallel_read_time;
    }

    Timespan parallel_write_time() const {
        return _parallel_write_time;
    }

    Timespan parallel_io_time() const {
        return _parallel_io_time;
    }

    Timespan elapsed_time() const {
        return _elapsed_time;
    }

    Timespan io_wait_time() const {
        return _io_wait_time;
    }

    Timespan read_wait_time() const {
        return _read_wait_time;
    }

    Timespan write_wait_time() const {
        return _write_wait_time;
    }
};
/*
std::ostream& operator << (std::ostream& o, const StatsData& s);

inline std::ostream& operator << (std::ostream& o, const Stats& s) {
    o << StatsData(s);
    return o;
}
*/
}

#endif
